package cn.lagou.sparkcore

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object homework {
  def main(args: Array[String]): Unit = {
    //定义sparkContext
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getCanonicalName)
    val sc = new SparkContext(conf)
    //设置日志级别
    sc.setLogLevel("warn")
    /**
     * 数据计算
     */
    //采用RDD：读取 http.log文件
    val httpData: RDD[Long] = sc.textFile("file:///F:\\code\\lagoubigdata\\data\\http.log")
      .map(x => ipConvert(x.split("\\|")(1)))

    //读取IP配置文件
    val ipData: Array[(Long, Long, String)] = sc.textFile("file:///F:\\code\\lagoubigdata\\data\\ip.dat")
      .map { line =>
        val field: Array[String] = line.split("\\|")
        (field(2).toLong, field(3).toLong, field(6))
      }.collect()
    val ipBC: Broadcast[Array[(Long, Long, String)]] = sc.broadcast(ipData.sortBy(_._1))

    //逐条数据比对，找到对应的城市。使用二分查找
    val results: Array[(Any, Int)] = httpData.mapPartitions { iter =>
      val ipsInfo: Array[(Long, Long, String)] = ipBC.value
      iter.map { ip =>
        val city: Any = getCityName(ip, ipsInfo)
        (city, 1)
      }
    }.reduceByKey(_ + _)
      .collect()
    results.sortBy(_._2)
      .foreach(println)

    //关闭连接
    sc.stop()

  }

  //定义一个IP转换函数 192.168.10.2  -- 192.168.25.3
  def ipConvert(ip: String): Long = {
    val arr: Array[Long] = ip.split("\\.")
      .map(_.toLong)
    var ipLong: Long = 0L
    for (i <- arr.indices) {
      val ans: Long = scala.math.pow(255, i).toLong
      ipLong += arr(i) * ans
    }
    ipLong
  }

  //寻找IP对应的城市
  def getCityName(ip: Long, ips: Array[(Long, Long, String)]): String = {
    var start = 0
    var end: Int = ips.length - 1
    var middle = 0

    while (start <= end) {
      middle = (start + end) / 2
      if ((ip >= ips(middle)._1) && (ip <= ips(middle)._2))
        return ips(middle)._3
      else if (ip < ips(middle)._1)
        end = middle - 1
      else
        start = middle + 1
    }
    "Unknown"
  }

}
